package com.yeming.flink.practice.transformation

import com.yeming.flink.practice.source.StationLog
import com.yeming.flink.practice.transformation.TranRickMapFunction.getClass
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object TranProcessFunction {

  // 监控所有的手机号码，如果这个手机号码在5秒内，所有呼叫它的日志都是失败的，则发出告警信息
  // 如果在5秒内只要有一个呼叫不是fail则不用告警
  def main(args: Array[String]): Unit = {
    //加载环境
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //读取数据源
    val stream: DataStream[String] = streamEnv.socketTextStream("f1",9999)
    //数据计算,
    val streamLog: DataStream[StationLog] = stream.map(line => {
      val arr: Array[String] = line.split(",")
      new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
    })

    val result: DataStream[String] = streamLog.keyBy(_.callIn)
      .process(new MonitorCallFail)
    result.print()

    streamEnv.execute()
  }

  class MonitorCallFail extends KeyedProcessFunction[String, StationLog, String] {
    // 使用状态对象记录时间
    lazy val timeState:ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("time",classOf[Long]))


    override def processElement(value: StationLog,
                                ctx: KeyedProcessFunction[String, StationLog, String]#Context,
                                out: Collector[String]): Unit = {
      //先从状态中取到时间
      var time = timeState.value()
      if (time == 0 && value.callType.equals("fail")) { //第一次发现呼叫失败
        //获取当前系统时间，并注册定时器
        val timeCurrent: Long = ctx.timerService().currentProcessingTime()
        //定时器在五秒后触发
        var onTime = timeCurrent + 5 * 1000L
        ctx.timerService().registerProcessingTimeTimer(onTime)
        //把触发的时间保存到状态中
        timeState.update(onTime)
      }
      if (time != 0 && !value.callType.equals("fail")) {
        ctx.timerService().deleteProcessingTimeTimer(time)
        timeState.clear()//清空当前状态
      }
    }

    override def onTimer(timestamp: Long,
                         ctx: KeyedProcessFunction[String, StationLog, String]#OnTimerContext,
                         out: Collector[String]): Unit = {
      var warnStr = "触发的时间：" + timestamp + ", 手机号：" + ctx.getCurrentKey
      out.collect(warnStr)
    }
  }

}
